package ex.tableapi;

import org.apache.flink.table.api.Table;

import static org.apache.flink.table.api.Expressions.$;

public class ExOrderBy extends   ApiFrame {
    private String outName = "printOutTable";

    public static void main(String[] args) {
        ExOrderBy demo = new ExOrderBy();
        demo.getEnv();
        Table orders = demo.registerMysql("myorder", "orders");

        Table result = demo.exOrderBy(orders);

        demo.registerConsole(demo.createPrintOutDDL(result.getResolvedSchema().toString()), demo.outName);
        result.executeInsert(demo.outName);
    }

    private Table exOrderBy(Table order1) {

        Table result = order1.orderBy($("rowtime").asc()).select($("a"), $("rowtime"));
        /**
         跟水印字段有关，新增数据时间应该是往后越来越大，小的数据会剔除
         */
        Table result1 = order1.orderBy($("rowtime").asc()).fetch(4);

// skips the first 3 records and returns all following records from the sorted result
        //TODO  FETCH is missed, which on streaming table is not supported currently.
//        Table result2 = order1.orderBy($("rowtime").asc()).offset(3);

// skips the first 10 records and returns the next 5 records from the sorted result
        Table result3 = order1.orderBy($("rowtime").asc()).offset(10).fetch(5);
        return result3;
    }

}
